package com.dap.configsvrclient.disconf.zk;

import java.io.IOException;
import java.text.MessageFormat;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;

import org.apache.curator.RetryPolicy;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.CuratorFrameworkFactory;
import org.apache.curator.framework.api.ACLBackgroundPathAndBytesable;
import org.apache.curator.framework.imps.CuratorFrameworkState;
import org.apache.curator.framework.recipes.cache.NodeCache;
import org.apache.curator.framework.recipes.cache.NodeCacheListener;
import org.apache.curator.retry.ExponentialBackoffRetry;
import org.apache.zookeeper.CreateMode;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.util.StringUtils;

import com.dap.configsvrclient.disconf.rest.DisconfRestMgr;
import com.dap.configsvrclient.util.ConfigSvrConfig;
import com.dap.core.context.ContextUtils;


public class ZkManager {
    private static final Logger LOGGER = LoggerFactory.getLogger(ZkManager.class);
    private static final ZkManager instance = new ZkManager();
    private static final String DISCONF_ZK_NODE = "{0}/{1}/{2}_{3}_{4}/file/";


    private ZkManager() {
        connect();
    }


    public static ZkManager getInstance() {
        return instance;
    }

    private CuratorFramework client = null;
    private final ExecutorService pool = Executors.newFixedThreadPool(10);


    public void connect() {
        connect(DisconfRestMgr.getInstance().retrieveZkHost());
    }


    public void connect(String host) {
        ExponentialBackoffRetry retryPolicy = new ExponentialBackoffRetry(1000, 3);

        this.client = connect(host, retryPolicy, Integer.MAX_VALUE, Integer.MAX_VALUE);
    }


    private CuratorFramework connect(String host, RetryPolicy retryPolicy, int connectionTimeoutMs,
            int sessionTimeoutMs) {
        if ((this.client == null) || (!CuratorFrameworkState.STARTED.equals(this.client.getState()))) {
            this.client =
                    CuratorFrameworkFactory.newClient(host, sessionTimeoutMs, connectionTimeoutMs,
                        retryPolicy);

            this.client.start();
        }
        return this.client;
    }


    public boolean create(String path, String node, boolean isPersistent) {
        checkClient();
        ACLBackgroundPathAndBytesable<String> b = null;
        if (isPersistent) {
            b = this.client.create().creatingParentsIfNeeded().withMode(CreateMode.PERSISTENT);
        } else {
            b = this.client.create().creatingParentsIfNeeded().withMode(CreateMode.EPHEMERAL);
        }
        try {
            if (!StringUtils.isEmpty(node)) {
                b.forPath(path, node.getBytes());
            } else {
                b.forPath(path);
            }
            return true;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
        }
        return false;
    }


    public boolean watch(String path, ZkNodeListener listener) {
        checkClient();

        final String key = path;
        final ZkNodeListener cb = listener;
        final NodeCache nodeCache = new NodeCache(this.client, path, false);
        try {
            nodeCache.start(true);
            nodeCache.getListenable().addListener(new NodeCacheListener() {
                @Override
                public void nodeChanged() throws Exception {
                    LOGGER.debug("config [{}] changed, new data is: [{}]", key, new String(nodeCache
                        .getCurrentData().getData()));
                    cb.update(ZkManager.this.getConfigKey(key));
                }
            }, this.pool);

            return true;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
            try {
                nodeCache.close();
            } catch (IOException ex) {
            }
        } finally {
            Runtime.getRuntime().addShutdownHook(new Thread(new Runnable() {
                public void run() {
                    try {
                        nodeCache.close();
                    } catch (IOException e) {
                    }
                }
            }));
        }

        return false;
    }


    public boolean isExists(String path) {
        checkClient();
        try {
            return this.client.checkExists().forPath(path) != null;
        } catch (Exception e) {
            LOGGER.error(e.toString(), e);
        }
        return false;
    }


    private void checkClient() {
        if ((this.client == null) || (!CuratorFrameworkState.STARTED.equals(this.client.getState()))) {
            throw new IllegalStateException("Please connect first!");
        }
    }


    public String assembleZkNodeKey(String configKey) {
        return disconfZkPathFormat() + configKey;
    }


    private String disconfZkPathFormat() {
        MessageFormat mf = new MessageFormat(DISCONF_ZK_NODE);
        return mf.format(new Object[] { DisconfRestMgr.getInstance().retrieveZkPrefix(),
                                       ContextUtils.getTenantId(),
                                       ConfigSvrConfig.getInstance().getAppName(),
                                       ConfigSvrConfig.getInstance().getVersion(),
                                       ConfigSvrConfig.getInstance().getEnv() });
    }


    public String getConfigKey(String zkNodeKey) {
        String tmp = disconfZkPathFormat();
        int idx = zkNodeKey.indexOf(tmp);
        if (idx != -1) {
            return zkNodeKey.substring(idx + tmp.length());
        }
        return zkNodeKey;
    }
}
